package com.corgi.message.kafka.serializer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.util.Map;

/**
 * @title: ObjectSerializer
 * @description: Object 序列化器
 * @author: dengmiao
 * @create: 2019-10-11 09:56
 **/
@Slf4j
public class ObjectSerializer implements Serializer<Object> {

	@Override
	public void configure(Map<String, ?> map, boolean b) {

	}

	@Override
	public byte[] serialize(String topic, Object object) {
		log.info("topic : {}, object : {}", topic, object);
		byte[] dataArray;
		ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
		try {
			ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
			objectOutputStream.writeObject(object);
			dataArray = outputStream.toByteArray();
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		return dataArray;
	}

	@Override
	public void close() {

	}
}
